package com.warren.financial.lease.realtime.util;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.warren.financial.lease.realtime.common.FinancialLeaseCommon;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @AUTHOR warren
 * @date 2023/12/11
 **/
public class CreateEnvUtil {

    public static StreamExecutionEnvironment getExecutionEnvironment(Integer port,String appName){

        //创建流环境
        Configuration conf = new Configuration();
        conf.setInteger(RestOptions.PORT,port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        // 设置检查点和状态后端
        env.enableCheckpointing(10 * 1000L);

        // 设置相邻的两个检查点最小的时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10 * 1000L);
        //停掉任务的时候，是否还需要保留检查点
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        // 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(FinancialLeaseCommon.HDFS_URI_PREFIX+appName);

        System.setProperty("HADOOP_USER_NAME",FinancialLeaseCommon.HADOOP_USER_NAME);

        return env;
    }

    public static MySqlSource<String> getMysqlSource(){
        MySqlSource<String> mySqlSourceBuilder = MySqlSource.<String>builder()
                .hostname(FinancialLeaseCommon.MYSQL_HOSTNAME)
                .port(3306)
                .databaseList(FinancialLeaseCommon.MEDICAL_CONFIG_DATABASE)
                .tableList(FinancialLeaseCommon.MEDICAL_CONFIG_TABLE)
                .username(FinancialLeaseCommon.MYSQL_USERNAME)
                .password(FinancialLeaseCommon.MYSQL_PASSWD)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        return mySqlSourceBuilder;
    }

}
